package com.tech.process

import com.tech.common.{KafkaUtil, KuduUtil, SimplyDateUtil}
import com.tech.config.ApplicationConfig
import com.tech.entity.Retailevent
import com.tech.process.nodeProcess.{BehaviorNode, NextNode, TestNode, WaitNode}
import org.json4s.DefaultFormats
import org.json4s.jackson.JsonMethods

import scala.collection.mutable
import scala.collection.mutable.ArrayBuffer
import scala.util.Random
import scala.util.control.Breaks
import scala.util.parsing.json.JSON

object EventProcess {

  def eventProcess(value: String,
                   topic: String,
                   activities: ArrayBuffer[String],
                   startNodeMap: mutable.HashMap[String, String],
                   nextNodeMap: mutable.HashMap[String, NextNode],
                   groupMap: mutable.HashMap[String, mutable.Set[String]],
                   behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
                   testNodeMap: mutable.HashMap[String, TestNode],
                   waitNodeMap: mutable.HashMap[String, WaitNode],
                   eventKeySet: mutable.HashSet[String],
                   dimDataGroupDetailMap: mutable.HashMap[String, ArrayBuffer[String]],
                   retaileventSchemaMap: mutable.HashMap[String, String],
                   columnTypeMap: mutable.HashMap[String, String]
                  ): Unit = {


    implicit val f: DefaultFormats.type = org.json4s.DefaultFormats

    val tmpMap = JsonMethods.parse(value).extract[Map[String, Any]]

    if (tmpMap("project") == ApplicationConfig.PROJECT && eventKeySet.contains(tmpMap("event_key").toString)) {


      val retaileventMap = mutable.HashMap[String, Any]()

      //将数据装载到可变的map中
      tmpMap.foreach(i => retaileventMap.put(i._1, i._2))

      if (retaileventMap.contains("customer_id") && retaileventMap("customer_id") != null && retaileventMap("customer_id") != "") {
        retaileventMap.put("is_member", "1")
      } else {
        //此处若有不包含detail_id脏数据会导致程序报错
        retaileventMap.put("customer_id", retaileventMap("detail_id"))
        retaileventMap.put("is_member", "0")
      }
      retaileventMap.put("event_month", SimplyDateUtil.getMonthFormat().format(retaileventMap("event_time")))
      if (!retaileventMap.contains("posted_time"))
        retaileventMap.put("posted_time", System.currentTimeMillis())

      val customer_id = retaileventMap("customer_id").toString
      val event_key = retaileventMap("event_key").toString
      val member_id = retaileventMap("member_id").toString
      val event_time = retaileventMap("event_time").toString.toLong

      activities.foreach(activity_id => {
        try {
        //查询用户是否参与活动
        if (KuduUtil.check(customer_id, activity_id)) {
          //查询用户当前节点中可匹配的节点(node = process_id)
          val nodes = KuduUtil.getNodes(customer_id, activity_id, event_key)
          if (nodes.nonEmpty)
            nodes.foreach(node => {
              println(customer_id + "  EventProcess-line-93:" + System.currentTimeMillis())
              if (event_key == behaviorNodeMap(node).event_type) {
                try {
                  filterRule(node, activity_id, retaileventMap, behaviorNodeMap, nextNodeMap, groupMap, testNodeMap, waitNodeMap, dimDataGroupDetailMap, retaileventSchemaMap, columnTypeMap)
                } catch {
                  case _: Exception => KafkaUtil.write2Kafka(s"errorMsg_$topic", s"activity_id:'$activity_id',process_id:'$node',customer_id:'$customer_id' => $value")
                }
              }
              println(customer_id + "  EventProcess-line-95:" + System.currentTimeMillis())
            })
        } else {
          println("未参与")
          println("activity_id:" + activity_id)
          val startNode = startNodeMap(activity_id)
          val nextIds = nextNodeMap(startNode).next_id
          if (nextIds != null)
            nextIds.split(",").foreach(
              next => {
                println("next:" + next)
                println(nextNodeMap(next).id_type == 100000005)
                if (nextNodeMap(next).id_type == 100000005) {
                  //非客群活动，用户未参与活动，需新建节点
                  KuduUtil.insertCustomerActivity(customer_id, activity_id)
                  //首次触达节点，记录节点结束时间
                  createDelayNode(customer_id, member_id, activity_id, next, behaviorNodeMap,event_time)
                  createNode(customer_id, activity_id, next, behaviorNodeMap, System.currentTimeMillis())
                  if (event_key == behaviorNodeMap(next).event_type) {
                    try {
                      filterRule(next, activity_id, retaileventMap, behaviorNodeMap, nextNodeMap, groupMap, testNodeMap, waitNodeMap, dimDataGroupDetailMap, retaileventSchemaMap, columnTypeMap)
                    } catch {
                      case _: Exception => KafkaUtil.write2Kafka(s"errorMsg_$topic",s"activity_id:'$activity_id',process_id:'$next',customer_id:'$customer_id' => $value")
                    }
                  }
                }
              })
        }
        } catch {
          case _: Exception => KafkaUtil.write2Kafka(s"errorMsg_$topic", s"activity_id:'$activity_id',customer_id:'$customer_id' => $value")
        }
      })

      println("end ==> " + event_key + "    " + customer_id + "     " + System.currentTimeMillis())

      retaileventMap.put("created_time", System.currentTimeMillis())

      KuduUtil.insertRow("retailevent", retaileventMap, retaileventSchemaMap)
    }
  }


  /**
   * 行为节点规则匹配
   *
   * @param process_id            节点id
   * @param activity_id           活动id
   * @param retaileventMap        retaileventMap
   * @param behaviorNodeMap       behaviorNodeMap
   * @param nextNodeMap           nextNodeMap
   * @param groupMap              groupMap
   * @param testNodeMap           testNodeMap
   * @param waitNodeMap           waitNodeMap
   * @param dimDataGroupDetailMap dimDataGroupDetailMap
   * @param retaileventSchemaMap  retaileventSchemaMap
   * @param columnTypeMap         columnTypeMap
   * @param isFirstVisit          是否首次匹配
   */
  def filterRule(process_id: String,
                 activity_id: String,
                 retaileventMap: mutable.HashMap[String, Any],
                 behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
                 nextNodeMap: mutable.HashMap[String, NextNode],
                 groupMap: mutable.HashMap[String, mutable.Set[String]],
                 testNodeMap: mutable.HashMap[String, TestNode],
                 waitNodeMap: mutable.HashMap[String, WaitNode],
                 dimDataGroupDetailMap: mutable.HashMap[String, ArrayBuffer[String]],
                 retaileventSchemaMap: mutable.HashMap[String, String],
                 columnTypeMap: mutable.HashMap[String, String],
                 isFirstVisit: Boolean = false): Unit = {

    val customer_id = retaileventMap("customer_id").toString
    val member_id = retaileventMap("member_id").toString
    val event_key = retaileventMap("event_key").toString
    val event_time = retaileventMap("event_time").toString.toLong

    val behaviorNode = behaviorNodeMap(process_id)
    val nextNode = nextNodeMap(process_id)

    val timeData = behaviorNode.time_condition
    // todo 通过json中的长度做判断，完全依赖mysql中该字段的规范写入，逻辑存在隐患
    if (timeData.length == 67) {

      println("timeData:" + timeData)
      val beginDate = SimplyDateUtil.getFormat().parse(timeData.substring(14, 33))
      val endDate = SimplyDateUtil.getFormat().parse(timeData.substring(46, 65))

      val beginTime = beginDate.getTime
      val endTime = endDate.getTime

      if (event_time >= beginTime && event_time <= endTime) {
        //匹配规则
        //行为属性判断
        var behaviorAttributeFlag = true
        if (behaviorNode.addition != null && behaviorNode.addition.contains("100000001")) {
          println("进入行为属性判断")
          behaviorAttributeFlag = matchBehaviorAttribute(behaviorNode.behavior_attribute, retaileventMap, dimDataGroupDetailMap, columnTypeMap)
          println(behaviorAttributeFlag)
        }


        //累计满足判断
        var measurePropertyFlag = true
        //做过：100000000 没做过：100000001
        if (behaviorNode.customer_behavior == 100000001) {
          if (behaviorAttributeFlag) {
            //确认不通过
            deleteNode(customer_id, activity_id, process_id, event_key, event_time)
          }
        } else if (behaviorAttributeFlag) {

          if (behaviorNode.addition != null && behaviorNode.addition.contains("100000000")) {
            println("162 进入累计满足判断")
            measurePropertyFlag = matchMeasureProperty(retaileventMap, behaviorNode, customer_id, activity_id, process_id, member_id, event_key, event_time, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap, isFirstVisit)
            println(measurePropertyFlag)
          } else {
            pass(retaileventMap, customer_id, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)
          }

          if (behaviorAttributeFlag && measurePropertyFlag) {
            //确认匹配，仍需将该节点软删除
            deleteNode(customer_id, activity_id, process_id, event_key, event_time)


          }
        }

      } else {
        //时间确认不匹配
        if (event_time > endTime)
          deleteNode(customer_id, activity_id, process_id, event_key, event_time)
      }

    } else {
      println("进入183行")
      val num = timeData.substring(9, timeData.length - 22).toInt
      val time_type = timeData.substring(timeData.length - 10, timeData.length - 1)
      val lastTime = {
        if (isFirstVisit) {
          System.currentTimeMillis()
        } else {
          KuduUtil.getNodeTime(customer_id, activity_id, nextNode.id)
        }
      }
      val endTime = time_type match {
        //天
        case "100000003" => lastTime + num.toLong * 1000 * 60 * 60 * 24
        //周
        case "100000002" => lastTime + num.toLong * 1000 * 60 * 60 * 24 * 7
        //月
        case "100000001" => lastTime + num.toLong * 1000 * 60 * 60 * 24 * 30
        //年
        case "100000000" => lastTime + num.toLong * 1000 * 60 * 60 * 24 * 365
        //未匹配，原样返回，时间必定小于event_time
        case _ => lastTime
      }

      if (event_time <= endTime) {
        println("node:" + process_id)
        //匹配规则
        //行为属性判断
        var behaviorAttributeFlag = true
        if (behaviorNode.addition != null && behaviorNode.addition.contains("100000001")) {
          behaviorAttributeFlag = matchBehaviorAttribute(behaviorNode.behavior_attribute, retaileventMap, dimDataGroupDetailMap, columnTypeMap)
        }

        println("behaviorAttributeFlag:" + behaviorAttributeFlag)
        println("behaviorNode.customer_behavior:" + behaviorNode.customer_behavior)
        println("behaviorNode.addition" + behaviorNode.addition)

        //累计满足判断
        var measurePropertyFlag = true
        //做过：100000000 没做过：100000001
        if (behaviorNode.customer_behavior == 100000001) {
          if (behaviorAttributeFlag) {
            //确认不通过
            deleteNode(customer_id, activity_id, process_id, event_key, event_time)
          }
        } else if (behaviorAttributeFlag) {

          if (behaviorNode.addition != null && behaviorNode.addition.contains("100000000")) {
            println("230 进入累计满足判断")
            measurePropertyFlag = matchMeasureProperty(retaileventMap, behaviorNode, customer_id, activity_id, process_id, member_id, event_key, event_time, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap, isFirstVisit)
          } else {
            pass(retaileventMap, customer_id, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)
          }

          if (behaviorAttributeFlag && measurePropertyFlag) {
            //确认匹配，仍需将该节点软删除
            deleteNode(customer_id, activity_id, process_id, event_key, event_time)

          }
        }

      } else {
        //时间确认不匹配
        deleteNode(customer_id, activity_id, process_id, event_key, event_time)
      }
    }


  }

  /**
   * 行为属性匹配
   *
   * @param behavior_attribute    行为属性条件
   * @param retaileventMap        当前事件
   * @param dimDataGroupDetailMap 标签群组map
   * @param columnTypeMap         列类型map
   * @return 是否匹配
   */
  def matchBehaviorAttribute( //customer_id:String,
                              behavior_attribute: String,
                              retaileventMap: mutable.HashMap[String, Any],
                              dimDataGroupDetailMap: mutable.HashMap[String, ArrayBuffer[String]],
                              columnTypeMap: mutable.HashMap[String, String]): Boolean = {

    var behaviorAttributeFlag = true
    val firstLoop = new Breaks
    val secondLoop = new Breaks

    //当前版本需求，所有判断均依赖当前消息，不需要获取用户信息
    //    val customerDetailResult = KuduUtil.getCustomerDetail(customer_id)
    val jsonObj = JSON.parseFull(behavior_attribute)
    val map: Map[String, Any] = jsonObj.get.asInstanceOf[Map[String, Any]]
    val firstFlag = map("firstLogicalVal").asInstanceOf[Double] == 0
    //firstLogicalVal 0:与 1:或
    if (firstFlag) {
      val list = map("firstCondition").asInstanceOf[List[Map[String, Any]]]
      //若存在一个不符合的条件则直接中断循环
      firstLoop.breakable {
        list.foreach(
          map1 => {
            var secondFlag = map1("secondLogicalVal").asInstanceOf[Double] == 0
            val list1 = map1("secondCondition").asInstanceOf[List[Map[String, Any]]]
            if (secondFlag) {
              secondLoop.breakable {
                list1.foreach(
                  map2 => {
                    if (!Retailevent.matchRule(map2, retaileventMap, dimDataGroupDetailMap, columnTypeMap)) {
                      behaviorAttributeFlag = !behaviorAttributeFlag
                      secondLoop.break()
                    }
                  }
                )
              }
            } else {
              behaviorAttributeFlag = !behaviorAttributeFlag
              secondLoop.breakable {
                list1.foreach(
                  map2 => {
                    if (Retailevent.matchRule(map2, retaileventMap, dimDataGroupDetailMap, columnTypeMap)) {
                      behaviorAttributeFlag = !behaviorAttributeFlag
                      secondLoop.break()
                    }
                  }
                )
              }

            }
            if (!behaviorAttributeFlag) {
              firstLoop.break()
            }
          }
        )
      }
    } else {
      behaviorAttributeFlag = !behaviorAttributeFlag
      val list = map("firstCondition").asInstanceOf[List[Map[String, Any]]]
      //若存在一个符合的条件则直接中断循环
      firstLoop.breakable {
        list.foreach(
          map1 => {
            val secondFlag = map1("secondLogicalVal").asInstanceOf[Double] == 0
            val list1 = map1("secondCondition").asInstanceOf[List[Map[String, Any]]]
            if (secondFlag) {
              secondLoop.breakable {
                list1.foreach(
                  map2 => {
                    if (!Retailevent.matchRule(map2, retaileventMap, dimDataGroupDetailMap, columnTypeMap)) {
                      secondLoop.break()
                    }
                  }
                )
                behaviorAttributeFlag = !behaviorAttributeFlag
              }
            } else {
              secondLoop.breakable {
                list1.foreach(
                  map2 => {
                    if (Retailevent.matchRule(map2, retaileventMap, dimDataGroupDetailMap, columnTypeMap)) {
                      behaviorAttributeFlag = !behaviorAttributeFlag
                      secondLoop.break()
                    }
                  }
                )
              }

            }
            if (behaviorAttributeFlag) {
              firstLoop.break()
            }
          }
        )
      }
    }

    behaviorAttributeFlag

  }

  /**
   * 累计满足匹配
   *
   * @param behaviorNode    行为节点
   * @param customer_id     用户id
   * @param activity_id     活动id
   * @param process_id      节点id
   * @param event_key       事件类型
   * @param event_time      事件时间
   * @param nextNodeMap     nextNodeMap
   * @param behaviorNodeMap behaviorNodeMap
   * @param testNodeMap     testNodeMap
   * @param waitNodeMap     waitNodeMap
   * @return 是否匹配
   */
  def matchMeasureProperty(retaileventMap: mutable.HashMap[String, Any],
                           behaviorNode: BehaviorNode,
                           customer_id: String,
                           activity_id: String,
                           process_id: String,
                           member_id: String,
                           event_key: String,
                           event_time: Long,
                           nextNodeMap: mutable.HashMap[String, NextNode],
                           behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
                           testNodeMap: mutable.HashMap[String, TestNode],
                           waitNodeMap: mutable.HashMap[String, WaitNode],
                           isFirstVisit: Boolean = false): Boolean = {

    //解析累计满足的json
    val json1Obj = JSON.parseFull(behaviorNode.measure_property)
    val map1 = json1Obj.get.asInstanceOf[Map[String, Any]]
    val limitSymbol = map1("limitSymbol").asInstanceOf[String]
    val limitName = map1("limitName").asInstanceOf[List[String]]

    var bool = false

    //除交易事件外，目前其他事件均只有count,目前仅有为count时List的长度为1
    if (limitName.length == 1) {

      bool = matchMeasurePropertyRule(retaileventMap, 1, limitSymbol, map1, customer_id, activity_id, process_id, member_id, event_key, event_time, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap, isFirstVisit)

    }
    //组合条件（例：销售额+平均值）
    else if (limitName.length == 2) {

      val eventSum = retaileventMap(limitName.head).toString.toDouble

      limitName(1) match {
        //最值 判断，不保存
        case "min" | "max" =>
          bool = matchMeasurePropertyRuleWithoutWrite(retaileventMap, eventSum, limitSymbol, map1, customer_id, activity_id, process_id, member_id, event_key, event_time, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap, isFirstVisit)
        //总和 判断，保存
        case "num" =>
          bool = matchMeasurePropertyRule(retaileventMap, eventSum, limitSymbol, map1, customer_id, activity_id, process_id, member_id, event_key, event_time, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap, isFirstVisit)
        //平均 不判断，保存
        case "beverage" =>
          val rowResult = KuduUtil.getNodeRow(customer_id, activity_id, process_id)

          val eventValue = if (!rowResult.isNull("event_value")) {
            rowResult.getDouble("event_value")
          } else {
            0
          }
          val eventNum = if (!rowResult.isNull("event_num")) {
            rowResult.getDouble("event_num")
          } else {
            0
          }
          updateNode(customer_id, activity_id, process_id, event_key, event_time, eventValue + eventSum, eventNum + 1)
      }


    }

    bool

  }

  def matchMeasurePropertyRule(retaileventMap: mutable.HashMap[String, Any],
                               currentValue: Double,
                               limitSymbol: String,
                               map1: Map[String, Any],
                               customer_id: String,
                               activity_id: String,
                               process_id: String,
                               member_id: String,
                               event_key: String,
                               event_time: Long,
                               nextNodeMap: mutable.HashMap[String, NextNode],
                               behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
                               testNodeMap: mutable.HashMap[String, TestNode],
                               waitNodeMap: mutable.HashMap[String, WaitNode],
                               isFirstVisit: Boolean = false): Boolean = {

    var bool = false

    if (limitSymbol.equals("between")) {
      val limitValue = map1("limitValue").asInstanceOf[String]
      val minValue = limitValue.split(",").head.toDouble
      val maxValue = limitValue.split(",")(1).toDouble
      if (currentValue > maxValue) {
        //确认不通过
        deleteNode(customer_id, activity_id, process_id, event_key, event_time)
      } else {
        if (isFirstVisit) {
          //无法确认，将累计满足的total写回kudu
          updateNode(customer_id, activity_id, process_id, event_key, event_time, currentValue, 1)
        } else {
          val rowResult = KuduUtil.getNodeRow(customer_id, activity_id, process_id)

          val eventValue = if (!rowResult.isNull("event_value")) {
            rowResult.getDouble("event_value")
          } else {
            0
          }
          val eventNum = if (!rowResult.isNull("event_num")) {
            rowResult.getDouble("event_num")
          } else {
            0
          }
          val total = eventValue + currentValue
          if (total > maxValue) {
            //确认不通过
            deleteNode(customer_id, activity_id, process_id, event_key, event_time)
          } else {
            //无法确认，将累计满足的total写回kudu
            updateNode(customer_id, activity_id, process_id, event_key, event_time, total, eventNum + 1)
          }
        }
      }
    } else {
      val limitValue = map1("limitValue").asInstanceOf[Double]
      limitSymbol match {
        //可以在节点结束前匹配成功的
        case "gt" | "gte" =>
          if (Retailevent.matchLimit(currentValue, limitValue, limitSymbol)) {
            //确认匹配，仍需将该节点软删除
            bool = true
            deleteNode(customer_id, activity_id, process_id, event_key, event_time)
            pass(retaileventMap, customer_id, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)
          } else {
            if (isFirstVisit) {
              //无法确认，将累计满足的total写回kudu
              updateNode(customer_id, activity_id, process_id, event_key, event_time, currentValue, 1)
            } else {
              val rowResult = KuduUtil.getNodeRow(customer_id, activity_id, process_id)

              val eventValue = if (!rowResult.isNull("event_value")) {
                rowResult.getDouble("event_value")
              } else {
                0
              }
              val eventNum = if (!rowResult.isNull("event_num")) {
                rowResult.getDouble("event_num")
              } else {
                0
              }
              val total = eventValue + currentValue
              if (Retailevent.matchLimit(total, limitValue, limitSymbol)) {
                //确认匹配，仍需将该节点软删除
                bool = true
                deleteNode(customer_id, activity_id, process_id, event_key, event_time)
                pass(retaileventMap, customer_id, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)
              } else {
                //无法确认，将累计满足的total写回kudu
                updateNode(customer_id, activity_id, process_id, event_key, event_time, total, eventNum + 1)
              }
            }
          }
        case "neq" =>
          if (currentValue > limitValue) {
            //确认匹配，仍需将该节点软删除
            bool = true
            deleteNode(customer_id, activity_id, process_id, event_key, event_time)
            pass(retaileventMap, customer_id, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)
          } else {
            if (isFirstVisit) {
              //无法确认，将累计满足的total写回kudu
              updateNode(customer_id, activity_id, process_id, event_key, event_time, currentValue, 1)
            } else {
              val rowResult = KuduUtil.getNodeRow(customer_id, activity_id, process_id)

              val eventValue = if (!rowResult.isNull("event_value")) {
                rowResult.getDouble("event_value")
              } else {
                0
              }
              val eventNum = if (!rowResult.isNull("event_num")) {
                rowResult.getDouble("event_num")
              } else {
                0
              }
              val total = eventValue + currentValue
              if (total > limitValue) {
                //确认匹配，仍需将该节点软删除
                bool = true
                deleteNode(customer_id, activity_id, process_id, event_key, event_time)
                pass(retaileventMap, customer_id, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)
              } else {
                //无法确认，将累计满足的total写回kudu
                updateNode(customer_id, activity_id, process_id, event_key, event_time, total, eventNum + 1)
              }
            }
          }
        //可以在节点结束前匹配失败的
        case "lt" | "lte" =>
          if (!Retailevent.matchLimit(currentValue, limitValue, limitSymbol)) {
            //确认不通过
            deleteNode(customer_id, activity_id, process_id, event_key, event_time)
          } else {
            if (isFirstVisit) {
              //无法确认，将累计满足的total写回kudu
              updateNode(customer_id, activity_id, process_id, event_key, event_time, currentValue, 1)
            } else {
              val rowResult = KuduUtil.getNodeRow(customer_id, activity_id, process_id)

              val eventValue = if (!rowResult.isNull("event_value")) {
                rowResult.getDouble("event_value")
              } else {
                0
              }
              val eventNum = if (!rowResult.isNull("event_num")) {
                rowResult.getDouble("event_num")
              } else {
                0
              }
              val total = eventValue + currentValue
              if (!Retailevent.matchLimit(currentValue, limitValue, limitSymbol)) {
                //确认不通过
                deleteNode(customer_id, activity_id, process_id, event_key, event_time)
              } else {
                //无法确认，将累计满足的total写回kudu
                updateNode(customer_id, activity_id, process_id, event_key, event_time, total, eventNum + 1)
              }
            }
          }
        case "eq" =>
          if (currentValue > limitValue) {
            //确认不通过
            deleteNode(customer_id, activity_id, process_id, event_key, event_time)
          } else {
            if (isFirstVisit) {
              //无法确认，将累计满足的total写回kudu
              updateNode(customer_id, activity_id, process_id, event_key, event_time, currentValue, 1)
            } else {
              val rowResult = KuduUtil.getNodeRow(customer_id, activity_id, process_id)

              val eventValue = if (!rowResult.isNull("event_value")) {
                rowResult.getDouble("event_value")
              } else {
                0
              }
              val eventNum = if (!rowResult.isNull("event_num")) {
                rowResult.getDouble("event_num")
              } else {
                0
              }
              val total = eventValue + currentValue
              if (total > limitValue) {
                //确认不通过
                deleteNode(customer_id, activity_id, process_id, event_key, event_time)
              } else {
                //无法确认，将累计满足的total写回kudu
                updateNode(customer_id, activity_id, process_id, event_key, event_time, total, eventNum + 1)
              }
            }
          }
      }
    }

    bool

  }

  def matchMeasurePropertyRuleWithoutWrite(retaileventMap: mutable.HashMap[String, Any],
                                           currentValue: Double,
                                           limitSymbol: String,
                                           map1: Map[String, Any],
                                           customer_id: String,
                                           activity_id: String,
                                           process_id: String,
                                           member_id: String,
                                           event_key: String,
                                           event_time: Long,
                                           nextNodeMap: mutable.HashMap[String, NextNode],
                                           behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
                                           testNodeMap: mutable.HashMap[String, TestNode],
                                           waitNodeMap: mutable.HashMap[String, WaitNode],
                                           isFirstVisit: Boolean = false): Boolean = {

    var bool = false

    if (limitSymbol.equals("between")) {
      val limitValue = map1("limitValue").asInstanceOf[String]
      val minValue = limitValue.split(",").head.toDouble
      val maxValue = limitValue.split(",")(1).toDouble
      if (currentValue > maxValue) {
        //确认不通过
        deleteNode(customer_id, activity_id, process_id, event_key, event_time)
      }
    } else {
      val limitValue = map1("limitValue").asInstanceOf[Double]
      limitSymbol match {
        //可以在节点结束前匹配成功的
        case "gt" | "gte" =>
          if (Retailevent.matchLimit(currentValue, limitValue, limitSymbol)) {
            //确认匹配，仍需将该节点软删除
            bool = true
            deleteNode(customer_id, activity_id, process_id, event_key, event_time)
            pass(retaileventMap, customer_id, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)
          }
        case "neq" =>
          if (currentValue > limitValue) {
            //确认匹配，仍需将该节点软删除
            bool = true
            deleteNode(customer_id, activity_id, process_id, event_key, event_time)
            pass(retaileventMap, customer_id, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap)
          }
        //可以在节点结束前匹配失败的
        case "lt" | "lte" =>
          if (!Retailevent.matchLimit(currentValue, limitValue, limitSymbol)) {
            //确认不通过
            deleteNode(customer_id, activity_id, process_id, event_key, event_time)
          }
        case "eq" =>
          if (currentValue > limitValue) {
            //确认不通过
            deleteNode(customer_id, activity_id, process_id, event_key, event_time)
          }
      }
    }

    bool

  }

  /**
   * 仅用于用户通过行为节点
   */
  def pass(retaileventMap: mutable.HashMap[String, Any],
           customer_id: String,
           activity_id: String,
           process_id: String,
           member_id: String,
           nextNodeMap: mutable.HashMap[String, NextNode],
           behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
           testNodeMap: mutable.HashMap[String, TestNode],
           waitNodeMap: mutable.HashMap[String, WaitNode]): Unit = {

    //    println("向kafka发送消息")
    //    KafkaUtil.send2KafkaAndWrite2Kudu(GlobalValue.customerNodeTopic, GlobalValue.project, activity_id, process_id, customer_id, member_id)
    KuduUtil.insertPassNodeMsg(customer_id, activity_id, process_id)

    println("开始向kudu存储行为记录")
    val event_key = retaileventMap("event_key").toString
    val event_time = retaileventMap("event_time").toString.toLong

    val source =
      try {
        retaileventMap("source").toString.toInt
      } catch {
        case _: Throwable => -1
      }
    val channel =
      try {
        retaileventMap("channel").toString.toInt
      } catch {
        case _: Throwable => -1
      }
    val correlation_type =
      try {
        retaileventMap("correlation_type").toString.toInt
      } catch {
        case _: Throwable => -1
      }

    val store =
      try {
        retaileventMap("store").toString
      } catch {
        case _: Throwable => null
      }

    val first_store =
      try {
        retaileventMap("first_store").toString
      } catch {
        case _: Throwable => null
      }

    val channelTuple = event_key match {
      case "$order" => (source, store)
      case "$signup" => (channel, first_store)
      case "$binding" => (correlation_type, first_store)
      case _ => (source, store)
    }

    KuduUtil.insertBehaviorRecord(event_key, activity_id, customer_id, process_id, member_id, channelTuple._1, channelTuple._2, event_time)
    println("完成向kudu存储行为记录")

    passNode(customer_id, activity_id, process_id, member_id, nextNodeMap, behaviorNodeMap, testNodeMap, waitNodeMap,event_time)
  }

  /**
   * 通过当前节点时调用
   *
   * @param customer_id     用户id
   * @param activity_id     活动id
   * @param node            节点id
   * @param nextNodeMap     nextNodeMap
   * @param behaviorNodeMap behaviorNodeMap
   * @param testNodeMap     testNodeMap
   * @param waitNodeMap     waitNodeMap
   */
  def passNode(customer_id: String,
               activity_id: String,
               node: String,
               member_id: String,
               nextNodeMap: mutable.HashMap[String, NextNode],
               //groupMap: mutable.HashMap[String, mutable.Set[String]],
               behaviorNodeMap: mutable.HashMap[String, BehaviorNode],
               testNodeMap: mutable.HashMap[String, TestNode],
               waitNodeMap: mutable.HashMap[String, WaitNode],
               event_time: Long = System.currentTimeMillis()): Unit = {

    /**
     * @param currentNode 当前节点
     */
    def toNextNode(currentNode: String): Unit = {
      val nextIds = nextNodeMap(currentNode).next_id

      if (nextIds != null)
        nextIds.split(",").foreach(
          next =>
            //100000000 开始
            //100000001 结束
            //100000002 用户
            //100000003 A/B Test
            //100000004 A/B Test Child
            //100000005 行为
            //100000006 奖励
            //100000007 等待
            //100000008 消息
            nextNodeMap(next).id_type match {

              case 100000001 =>
              case 100000002 =>
              //                //判断是否属于该客群
              //                if (groupMap.contains(customer_id)) {
              //                  toNextNode(next)
              //                }
              case 100000003 =>
                //                KafkaUtil.send2KafkaAndWrite2Kudu(GlobalValue.customerNodeTopic, GlobalValue.project, activity_id, next, customer_id, member_id)
                KuduUtil.insertPassNodeMsg(customer_id, activity_id, next)
                val loop = new Breaks
                val random = new Random()
                val pro = random.nextInt(100)
                var tmp = 0
                loop.breakable {
                  nextNodeMap(next).next_id.split(",").foreach(
                    tmpNode => {
                      tmp += testNodeMap(tmpNode).probability
                      if (tmp > pro) {
                        //                        KafkaUtil.send2KafkaAndWrite2Kudu(GlobalValue.customerNodeTopic, GlobalValue.project, activity_id, tmpNode, customer_id, member_id)
                        KuduUtil.insertPassNodeMsg(customer_id, activity_id, tmpNode)
                        //做random分流
                        toNextNode(tmpNode)
                        loop.break()
                      }
                    }
                  )
                }
              case 100000004 =>
                //                KafkaUtil.send2KafkaAndWrite2Kudu(GlobalValue.customerNodeTopic, GlobalValue.project, activity_id, next, customer_id, member_id)
                KuduUtil.insertPassNodeMsg(customer_id, activity_id, next)
                //直接通过
                toNextNode(next)
              case 100000005 =>
                createDelayNode(customer_id, member_id, activity_id, next, behaviorNodeMap)
                createNode(customer_id, activity_id, next, behaviorNodeMap, System.currentTimeMillis())
              case 100000006 =>
                //发送消息到kafka
                println("奖励")
                KafkaUtil.send2Kafka(ApplicationConfig.CUSTOMER_NODE_TOPIC, ApplicationConfig.PROJECT, activity_id, next, customer_id, member_id)
                KuduUtil.insertPassNodeMsg(customer_id, activity_id, next)
                toNextNode(next)
              case 100000007 =>
                //发送消息到延时事件表
                val action_time = if (waitNodeMap(next).wait_type == 100000000) {
                  event_time + (waitNodeMap(next).unit match {
                    case 100000001 => waitNodeMap(next).wait_duration.toLong * 1000 * 60 * 60 * 24 * 30
                    case 100000002 => waitNodeMap(next).wait_duration.toLong * 1000 * 60 * 60 * 24 * 7
                    case 100000003 => waitNodeMap(next).wait_duration.toLong * 1000 * 60 * 60 * 24
                    case 100000004 => waitNodeMap(next).wait_duration.toLong * 1000 * 60 * 60
                  })
                } else {
                  waitNodeMap(next).wait_date * 1000
                }
                KuduUtil.insertDelayNode(ApplicationConfig.PROJECT, customer_id, member_id, activity_id, next, action_time, "wait", is_group = "0")
              case 100000008 =>
                //发送消息到kafka
                println("消息")
                KafkaUtil.send2Kafka(ApplicationConfig.CUSTOMER_NODE_TOPIC, ApplicationConfig.PROJECT, activity_id, next, customer_id, member_id)
                KuduUtil.insertPassNodeMsg(customer_id, activity_id, next)
                toNextNode(next)

            }
        )

    }

    toNextNode(node)

  }

  /**
   * 在用户节点表中新建数据
   *
   * @param customer_id     用户id
   * @param activity_id     活动id
   * @param process_id      节点id
   * @param behaviorNodeMap 行为节点规则Map
   * @param create_time     创建时间
   */
  def createNode(customer_id: String, activity_id: String, process_id: String, behaviorNodeMap: mutable.HashMap[String, BehaviorNode], create_time: Long): Unit = {

    val event_key = behaviorNodeMap(process_id).event_type
    //判断是否节点包含累计满足的规则
    if (behaviorNodeMap(process_id).addition != null && behaviorNodeMap(process_id).addition.contains("100000000")) {

      implicit val f: DefaultFormats.type = org.json4s.DefaultFormats
      val mpMap = JsonMethods.parse(behaviorNodeMap(process_id).measure_property).extract[Map[String, Any]]

      val limitSymbol = mpMap("limitSymbol").asInstanceOf[String]
      val limitName = mpMap("limitName").asInstanceOf[List[String]]
      if (limitSymbol.equals("between")) {
        val limitValue = mpMap("limitValue").asInstanceOf[String]
        val minValue = limitValue.split(",").head.toDouble
        val maxValue = limitValue.split(",")(1).toDouble
        if (limitName.length == 1) {
          KuduUtil.upsertNode(customer_id, activity_id, process_id, event_key, null, create_time, limitName.head, limitSymbol,
            limit_value_min = minValue, limit_value_max = maxValue, event_value = 0, event_num = 0)
        } else {
          KuduUtil.upsertNode(customer_id, activity_id, process_id, event_key, null, create_time, limitName.head + "|" + limitName(1), limitSymbol,
            limit_value_min = minValue, limit_value_max = maxValue, event_value = 0, event_num = 0)
        }
      } else {
        val limitValue = mpMap("limitValue").toString.toDouble
        if (limitName.length == 1) {
          KuduUtil.upsertNode(customer_id, activity_id, process_id, event_key, null, create_time, limitName.head, limitSymbol,
            limit_value = limitValue, event_value = 0, event_num = 0)
        } else {
          KuduUtil.upsertNode(customer_id, activity_id, process_id, event_key, null, create_time, limitName.head + "|" + limitName(1), limitSymbol,
            limit_value = limitValue, event_value = 0, event_num = 0)
        }
      }

    } else {
      KuduUtil.upsertNode(customer_id, activity_id, process_id, event_key, null, create_time)
    }
  }

  /**
   * 在用户节点表中更新数据（仅在触发规则匹配时触发）
   *
   * @param customer_id 用户id
   * @param activity_id 活动id
   * @param process_id  节点id
   * @param create_time 创建时间
   *
   */
  def updateNode(customer_id: String,
                 activity_id: String,
                 process_id: String,
                 event_key: String,
                 create_time: Long,
                 event_value: Double,
                 event_num: Double): Unit = {

    KuduUtil.upsertNode(customer_id, activity_id, process_id, event_key, null, create_time, event_value = event_value, event_num = event_num)

  }

  def deleteNode(customer_id: String, activity_id: String, process_id: String, event_key: String, create_time: Long): Unit = {

    KuduUtil.upsertNode(customer_id, activity_id, process_id, event_key, "1", create_time)

  }

  /**
   * 向延时节点表新增数据
   *
   * @param customer_id     用户id
   * @param activity_id     活动id
   * @param process_id      节点id
   * @param behaviorNodeMap 行为节点map
   */
  def createDelayNode(customer_id: String, member_id: String, activity_id: String, process_id: String, behaviorNodeMap: mutable.HashMap[String, BehaviorNode], event_time: Long = System.currentTimeMillis()): Unit = {

    implicit val f: DefaultFormats.type = org.json4s.DefaultFormats
    val timeMap = JsonMethods.parse(behaviorNodeMap(process_id).time_condition).extract[Map[String, Any]]
    println("timeMap:" + timeMap)
    val action_time = if (timeMap.contains("endTime")) {
      SimplyDateUtil.getFormat().parse(timeMap("endTime").toString).getTime
    } else {
      val count = timeMap("count").toString.toDouble.toInt
      //100000000年 100000001月 100000002周 100000003天
      event_time + (timeMap("unitTime").toString.toDouble.toInt match {
        case 100000000 => count.toLong * 1000 * 60 * 60 * 24 * 365
        case 100000001 => count.toLong * 1000 * 60 * 60 * 24 * 30
        case 100000002 => count.toLong * 1000 * 60 * 60 * 24 * 7
        case 100000003 => count.toLong * 1000 * 60 * 60 * 24
      })
    }
    //判断是否节点包含累计满足的规则
    if (behaviorNodeMap(process_id).addition != null && behaviorNodeMap(process_id).addition.contains("100000000")) {

      val mpMap = JsonMethods.parse(behaviorNodeMap(process_id).measure_property).extract[Map[String, Any]]

      val limitSymbol = mpMap("limitSymbol").toString
      if (limitSymbol != "between") {
        val limitValue = mpMap("limitValue").toString.toDouble
        KuduUtil.insertDelayNode(ApplicationConfig.PROJECT, customer_id, member_id, activity_id, process_id, action_time, "behavior", behaviorNodeMap(process_id).event_type, 0, limitValue, limit_symbol = limitSymbol, is_group = "0")
      } else {
        val limitValue = mpMap("limitValue").toString
        val minLimitValue = limitValue.split(",").head.toDouble
        val maxLimitValue = limitValue.split(",")(1).toDouble
        KuduUtil.insertDelayNode(ApplicationConfig.PROJECT, customer_id, member_id, activity_id, process_id, action_time, "behavior", behaviorNodeMap(process_id).event_type, 0, limit_value_min = minLimitValue, limit_value_max = maxLimitValue, limit_symbol = limitSymbol, is_group = "0")

      }
    } else {
      KuduUtil.insertDelayNode(ApplicationConfig.PROJECT, customer_id, member_id, activity_id, process_id, action_time, "behavior", behaviorNodeMap(process_id).event_type, is_group = "0")

    }
  }


}

